package com.atguigu.day03;

import com.atguigu.beans.WaterSensor;
import com.atguigu.func.WaterSensorMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Felix
 * @date 2024/4/1
 * 该案例演示了聚合算子---max\maxBy
 * 需求：统计不同的传感器采集到的水位最大值
 */
public class Flink13_Trans_Max_Maxby {
    public static void main(String[] args) throws Exception {
        //TODO 1.指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //TODO 2.准备数据
        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 8888);
        //TODO 3.将读取的数据  封装为一个WaterSensor对象
        SingleOutputStreamOperator<WaterSensor> wsDS = socketDS.map(
                new WaterSensorMapFunction()
        );
        //TODO 4.按照传感器id进行分组    经过keyby分组后得到的流叫分组流或者键控流
        KeyedStream<WaterSensor, String> keyedDS = wsDS.keyBy(WaterSensor::getId);
        //TODO 5.聚合---取出最大值
        //在Flink的聚合算子中，会将中间聚合的结果保存起来，保存在"状态"中
        //27> WaterSensor{id='ws1', ts=10, vc=10}
        //27> WaterSensor{id='ws1', ts=10, vc=20}
        //SingleOutputStreamOperator<WaterSensor> sumDS = keyedDS.max("vc");
        SingleOutputStreamOperator<WaterSensor> sumDS = keyedDS.maxBy("vc");
        //TODO 6. 打印
        sumDS.print();
        //TODO 7.提交作业
        env.execute();
    }
}
